package com.atguigu.edu.realtime.app.dws.user;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.PropertyNamingStrategy;
import com.alibaba.fastjson.serializer.SerializeConfig;
import com.atguigu.edu.realtime.app.BaseAppV2;
import com.atguigu.edu.realtime.bean.UserActionBean;
import com.atguigu.edu.realtime.common.Constant;
import com.atguigu.edu.realtime.common.KafkaTopicConfig;
import com.atguigu.edu.realtime.util.DateFormatUtil;
import com.atguigu.edu.realtime.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.HashMap;

public class DwsUserActionWindow extends BaseAppV2 {
    public static void main(String[] args) {
        new DwsUserActionWindow().init(3081,
                2,
                "DwsUserActionWindow",
                Constant.TOPIC_DWD_USER_CART_ADD,
                Constant.TOPIC_DWD_TRADE_ORDER_DETAIL,
                KafkaTopicConfig.DWD_TRAFFIC_PAGE_LOG_TOPIC,
                Constant.TOPIC_DWD_TRADE_PAY_SUC
                );
    }
    @Override
    protected void handle(StreamExecutionEnvironment env, HashMap<String, DataStreamSource<String>> streams) {
        //转换各个流，统一成bean,union成一个流
        DataStream<UserActionBean> beanStream = parseUnion(streams);
        //beanStream.print();
        //开窗聚合  注意迟到数据问题
        SingleOutputStreamOperator<UserActionBean> result = winAgg(beanStream);
        //写出到Doris
        writeToDoris(result);
    }

    private void writeToDoris(SingleOutputStreamOperator<UserActionBean> result) {
        result
                .map(bean -> {
                    SerializeConfig conf = new SerializeConfig();
                    conf.propertyNamingStrategy = PropertyNamingStrategy.SnakeCase;
                    return JSON.toJSONString(bean, conf);
                }).addSink(FlinkSinkUtil.getDorisSink("edu.dws_user_action_window"));
    }

    private SingleOutputStreamOperator<UserActionBean> winAgg(DataStream<UserActionBean> beanStream) {
        return beanStream
                .assignTimestampsAndWatermarks(WatermarkStrategy
                        .<UserActionBean>forBoundedOutOfOrderness(Duration.ofSeconds(15))
                        .withTimestampAssigner((bean,ts)->bean.getTs())
                )
                .windowAll(TumblingEventTimeWindows.of(Time.seconds(5)))
                .reduce(new ReduceFunction<UserActionBean>() {
                    @Override
                    public UserActionBean reduce(UserActionBean b1, UserActionBean b2) throws Exception {
                        b1.setHomeUvCt(b1.getHomeUvCt() + b2.getHomeUvCt());
                        b1.setCourseDetailUvCt(b1.getCourseDetailUvCt() + b2.getCourseDetailUvCt());
                        b1.setCartAddCt(b1.getCartAddCt() + b2.getCartAddCt());
                        b1.setOrderUuCt(b1.getOrderUuCt() + b2.getOrderUuCt());
                        b1.setPaySucUuCt(b1.getPaySucUuCt() + b2.getPaySucUuCt());

                        return b1;
                    }
                }, new ProcessAllWindowFunction<UserActionBean, UserActionBean, TimeWindow>() {
                    @Override
                    public void process(Context ctx, Iterable<UserActionBean> iterable, Collector<UserActionBean> out) throws Exception {
                        UserActionBean bean = iterable.iterator().next();
                        bean.setStt(DateFormatUtil.toDateTimeString(ctx.window().getStart()));
                        bean.setEdt(DateFormatUtil.toDateTimeString(ctx.window().getEnd()));
                        bean.setCurDate(DateFormatUtil.toDateString(ctx.window().getStart()));

                        out.collect(bean);
                    }
                });
    }

    private DataStream<UserActionBean> parseUnion(HashMap<String, DataStreamSource<String>> streams) {
        SingleOutputStreamOperator<UserActionBean> homeAndDetailStream = streams
                .get(KafkaTopicConfig.DWD_TRAFFIC_PAGE_LOG_TOPIC)
                .map(JSON::parseObject)
                .filter(obj -> {
                    String pageId = obj.getJSONObject("page").getString("page_id");
                    return "home".equals(pageId) || "course_detail".equals(pageId);
                })
                .keyBy(obj -> obj.getJSONObject("common").getString("uid"))
                .process(new KeyedProcessFunction<String, JSONObject, UserActionBean>() {

                    private ValueState<String> detailDateState;
                    private ValueState<String> homeDateState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        homeDateState = getRuntimeContext().getState(new ValueStateDescriptor<String>("homeDateState", String.class));
                        detailDateState = getRuntimeContext().getState(new ValueStateDescriptor<String>("detailDateState", String.class));
                    }

                    @Override
                    public void processElement(JSONObject obj, Context ctx, Collector<UserActionBean> out) throws Exception {
                        String pageId = obj.getJSONObject("page").getString("page_id");
                        Long ts = obj.getLong("ts");
                        String currDate = DateFormatUtil.toDateString(ts);

                        //找到第一次浏览首页 详情页 记录

                        Long homeUvCt = 0L;
                        Long courseDetailUvCt = 0L;

                        if ("home".equals(pageId) && !currDate.equals(homeDateState.value())) {
                            homeUvCt = 1L;
                            homeDateState.update(currDate);
                        } else if ("course_detail".equals(pageId) && !currDate.equals(detailDateState.value())) {
                            courseDetailUvCt = 1L;
                            detailDateState.update(currDate);
                        }

                        if (homeUvCt + courseDetailUvCt == 1) {
                            out.collect(new UserActionBean("", "", "",
                                    homeUvCt, courseDetailUvCt, 0L, 0L, 0L, ts));
                        }
                    }
                });

        SingleOutputStreamOperator<UserActionBean> cartAddStream = streams
                .get(Constant.TOPIC_DWD_USER_CART_ADD)
                .map(JSON::parseObject)
                .keyBy(obj -> obj.getString("user_id"))
                .process(new KeyedProcessFunction<String, JSONObject, UserActionBean>() {

                    private ValueState<String> cartAddState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        cartAddState = getRuntimeContext().getState(new ValueStateDescriptor<String>("cartAddState", String.class));
                    }

                    @Override
                    public void processElement(JSONObject obj, Context ctx, Collector<UserActionBean> out) throws Exception {
                        String lastDate = cartAddState.value();
                        Long ts = obj.getLong("ts");
                        String currDate = DateFormatUtil.toDateString(ts);

                        if(!currDate.equals(lastDate)){
                            cartAddState.update(currDate);

                            out.collect(new UserActionBean("", "", "",
                                    0L, 0L, 1L, 0L, 0L, ts));
                        }

                    }
                });


        SingleOutputStreamOperator<UserActionBean> orderStream = streams
                .get(Constant.TOPIC_DWD_TRADE_ORDER_DETAIL)
                .map(JSON::parseObject)
                .keyBy(obj -> obj.getString("user_id"))
                .process(new KeyedProcessFunction<String, JSONObject, UserActionBean>() {

                    private ValueState<String> orderState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        orderState = getRuntimeContext().getState(new ValueStateDescriptor<String>("orderState", String.class));
                    }

                    @Override
                    public void processElement(JSONObject obj, Context ctx, Collector<UserActionBean> out) throws Exception {
                        String lastDate = orderState.value();
                        Long ts = obj.getLong("oi_ts")*1000;
//                        System.out.println(ts);
                        String currDate = DateFormatUtil.toDateString(ts);

                        if (!currDate.equals(lastDate)) {
                            orderState.update(currDate);

                            out.collect(new UserActionBean("", "", "",
                                    0L, 0L, 0L, 1L, 0L, ts));
                        }
                    }
                });

        SingleOutputStreamOperator<UserActionBean> paySucStream = streams
                .get(Constant.TOPIC_DWD_TRADE_PAY_SUC)
                .map(JSON::parseObject)
                .keyBy(obj -> obj.getString("user_id"))
                .process(new KeyedProcessFunction<String, JSONObject, UserActionBean>() {

                    private ValueState<String> payState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        payState = getRuntimeContext().getState(new ValueStateDescriptor<String>("payState", String.class));
                    }

                    @Override
                    public void processElement(JSONObject obj, Context ctx, Collector<UserActionBean> out) throws Exception {
                        String lastDate = payState.value();
                        Long ts = obj.getLong("ts");
                        String currDate = DateFormatUtil.toDateString(ts);

                        if (!currDate.equals(lastDate)) {
                            payState.update(currDate);

                            out.collect(new UserActionBean("", "", "",
                                    0L, 0L, 0L, 0L, 1L, ts));
                        }
                    }
                });
        return homeAndDetailStream.union(cartAddStream,orderStream,paySucStream);
    }
}
